package cn.ac.iscas.service.sink;

import cn.ac.iscas.base.utils.*;
import cn.ac.iscas.config.MiddlePlatformConfig;
import cn.ac.iscas.domain.response.middleplatform.MetaColumnDTO;
import cn.ac.iscas.domain.response.middleplatform.MetaTableDTO;
import cn.ac.iscas.domain.response.middleplatform.ResponseEntity;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.queue.flink.FlinkTableQueueService;
import com.iscas.datasong.lib.common.DataSongException;
import com.iscas.datasong.lib.util.DataSongJsonUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static cn.ac.iscas.base.constant.JdbcConnectorConstant.*;

/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/8/10 15:06
 */
public class MySqlSink {
    private final static String ID = "mysql";
    private final static String CONNECTOR_DRIVER_VALUE = "com.mysql.jdbc.Driver";
    private final static String CONNECTOR_TYPE_VALUE = "jdbc";
    private OkHttpClientUtils httpClient = new OkHttpClientUtils();
    private String jobId;
    private String componentId;
    private String currentTableName;
    private List<FlinkSourceColumn> currentColumns;

    private String frontTableName;
    private List<FlinkSourceColumn> frontColumns;


    public MySqlSink(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public MySqlSink(String jobId, String taskId, String frontTableName, List<FlinkSourceColumn> frontColumns) {
        this.jobId = jobId;
        this.componentId = taskId;
        this.frontTableName = frontTableName;
        this.frontColumns = frontColumns;
        this.currentColumns = frontColumns;
    }


    public MySqlSink() {

    }


    public List<String> getSql(Map<String, Object> parameters) {
        String host = ParameterUtils.getRequiredString(parameters, "host");
        String port = String.valueOf(ParameterUtils.getRequiredInt(parameters, "port"));
        String db = ParameterUtils.getRequiredString(parameters, "databaseName");
        String table = ParameterUtils.getRequiredString(parameters, "table");
        String tableWithCatalog = String.format("%s.%s.%s", jobId, jobId, table);
        this.currentTableName = tableWithCatalog;
        String datasource = ParameterUtils.getRequiredString(parameters, "datasourceName");
        String username = "";
        String password = "";
        try {
            username = AesUtils.aesDecrypt(ParameterUtils.getRequiredString(parameters, "username"));
            password = AesUtils.aesDecrypt(ParameterUtils.getRequiredString(parameters, "password"));
        } catch (Exception e) {
            e.printStackTrace();
        }

        String jdbcUrl = new StringBuffer("jdbc:mysql://").append(host).append(":").append(port).append("/").append(db).toString().trim() + "?useUnicode=true&characterEncoding=UTF-8";
        StringBuffer withBuf = new StringBuffer("WITH (")
                .append(String.format("'%s' = '%s'", CONNECTOR_TYPE_KEY, CONNECTOR_TYPE_VALUE))
                .append(String.format(",'%s' = '%s'", CONNECTOR_URL_KEY, jdbcUrl))
                .append(String.format(",'%s' = '%s'", CONNECTOR_TABLE_KEY, table))
                .append(String.format(",'%s' = '%s'", CONNECTOR_USERNAME_KEY, username))
                .append(String.format(",'%s' = '%s'", CONNECTOR_PASSWORD_KEY, password));

        String withStr = withBuf.append(")").toString();
        StringBuffer tableSchemaBuf = new StringBuffer(String.format("CREATE TABLE %s (", tableWithCatalog));

        MiddlePlatformConfig middlePlatformConfig = SpringUtil.getBean(MiddlePlatformConfig.class);
        String tableMetaUrl = middlePlatformConfig.getQueryDataSourceTablemetaUrl() + datasource;
        ResponseEntity<MetaTableDTO> response = null;
        try {

            String tableMetaStr = httpClient.doGet(tableMetaUrl);
            response = DataSongJsonUtils.fromJson(tableMetaStr, ResponseEntity.class);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (DataSongException e) {
            e.printStackTrace();
        }
        List<Map<String, Object>> metaTables = (List<Map<String, Object>>) response.getValue();
        List<String> primaryKeys = new ArrayList<>();
        List<FlinkSourceColumn> columns = new ArrayList<>();
        for (Map<String, Object> metaTable : metaTables) {
            if (metaTable.getOrDefault("tableName", "").toString().equalsIgnoreCase(table)) {
                MetaTableDTO metaTableDTO = null;
                try {
                    metaTableDTO = DataSongJsonUtils.fromJson(metaTable.getOrDefault("tableMeta", "{}").toString(), MetaTableDTO.class);
                } catch (DataSongException e) {
                    e.printStackTrace();
                }

                for (MetaColumnDTO column : metaTableDTO.getMetaColumns()) {
                    tableSchemaBuf.append(String.format("%s %s, ", column.getName(), TypeConvertUtil.mysqlTypeToFlinkTypeString(column.getColumnType())));
                    if (column.isPrimary()) {
                        primaryKeys.add(column.getName());
                    }

                    FlinkSourceColumn flinkSourceColumn = new FlinkSourceColumn();
                    flinkSourceColumn.setName(column.getName());
                    flinkSourceColumn.setType(TypeConvertUtil.mysqlTypeToFlinkTypeString(column.getColumnType()));
                    flinkSourceColumn.setPrimaryKey(column.isPrimary());
                    flinkSourceColumn.setDefaultVal(column.getDefaultVal());
                    columns.add(flinkSourceColumn);
                }
            }
        }

        this.currentColumns = columns;


        if (!primaryKeys.isEmpty()) {
            tableSchemaBuf.append(String.format(" PRIMARY KEY (%s) NOT ENFORCED ", String.join(",", primaryKeys)));
        }

        String tableSchemaString = tableSchemaBuf.toString();
        if (tableSchemaString.trim().endsWith(",")) {
            tableSchemaString = tableSchemaString.substring(0, tableSchemaString.lastIndexOf(","));
        }

        tableSchemaString += ") ";

        String createTableSql = tableSchemaString + withStr;

        String insertSql = String.format("INSERT INTO %s ( SELECT * FROM %s )", tableWithCatalog, frontTableName);
        List<String> sqls = new ArrayList<>();
        sqls.add(createTableSql);
        sqls.add(insertSql);
        updateSource();
        return sqls;
    }

    public void updateSource() {
        FlinkTableQueueService flinkTableQueueService = SpringUtil.getBean(FlinkTableQueueService.class);
        Map<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
        tableColumns.put(this.currentTableName, this.currentColumns);
        flinkTableQueueService.put(this.jobId, this.componentId, tableColumns);
    }

    public static void main(String[] args) {
        Map<String, Object> params = new HashMap<>(32);
        params.put("ip", "172.16.10.159");
        params.put("port", "3306");
        params.put("db", "db_test_01");
        params.put("table", "ods_test_01");
        params.put("username", "2EPY14iZK5RDFsmzEdffJQ==");//root
        params.put("password", "2EPY14iZK5RDFsmzEdffJQ==");
        params.put("datasource", "db_test_01");

//        MySqlSource mySqlSource = new MySqlSource();
//        mySqlSource.getSql(params);


    }
}
